
package com.lmx.local;

import java.io.IOException;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.messaging.simp.stomp.StompFrameHandler;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.sockjs.client.SockJsClient;
import org.springframework.web.socket.sockjs.client.Transport;
import org.springframework.web.socket.sockjs.client.WebSocketTransport;

import com.lmx.common.util.PropertiesFileUtil;
import com.lmx.local.business.VideoProcessor;
import com.lmx.repoDomain.tmodel.LsMessage;

import net.sf.json.JSONObject;

/**
 * 本地server连接云端的socket类
 * 
 * @author 梁慕学
 * @version 1.00
 */
@Service
@Scope("singleton")
public class SocketClient {
    private static final Logger log = LoggerFactory.getLogger(SocketClient.class);
    
    private Properties properties = PropertiesFileUtil.getApplicationProperties();
    
	private final ExecutorService pool = Executors.newFixedThreadPool(10);
    
    @Autowired
    private VideoProcessor videoProcessor;
    
    WebSocketSession session;
    String topicChannel = "localServer";
    String conFlag="none";
    /*
     * 连接云端服务器
     */
    public void connect() throws InterruptedException, ExecutionException {
//        String uri = properties.getProperty("local.socketServerUri");
//        WebSocketClient simpleWebSocketClient = new StandardWebSocketClient();
//        List<Transport> transports = new ArrayList<>(1);
//        transports.add(new WebSocketTransport(simpleWebSocketClient));
//        SockJsClient sockJsClient = new SockJsClient(transports);
//        WebSocketStompClient stompClient = new WebSocketStompClient(sockJsClient);
//        stompClient.setMessageConverter(new MappingJackson2MessageConverter());
//        String tenantCode = properties.getProperty("local.tenantCode");
//        //定义处理方式
//        StompSessionHandler sessionHandler = new MyStompSessionHandler(tenantCode);
//        String endpointStr = uri + "/" + this.topicChannel;
//        //进行握手连接
//        session = stompClient.connect(uri, sessionHandler).get();
//        //发送socket登录请求
//        LsMessage msg = new LsMessage();
//        msg.setCode("01");
//        msg.setTenantCode(tenantCode);
//        session.send(topicChannel, msg);
    }
    
    /*
     * 连接云端服务器
     */
    public void connect2() throws InterruptedException, ExecutionException {
        String uri = properties.getProperty("local.socketServerUri");
        WebSocketClient simpleWebSocketClient = new StandardWebSocketClient();
        List<Transport> transports = new ArrayList<>(1);
        transports.add(new WebSocketTransport(simpleWebSocketClient));
        SockJsClient sockJsClient = new SockJsClient(transports);
        String endpointStr = uri + "/" + this.topicChannel;
        log.debug("endpointStr:" + endpointStr);
        sockJsClient.doHandshake(new MyWebSocketHandler(), endpointStr);
    }
  
    /*
     * 通用的发送机制
     */
    public void sendMessage(LsMessage msg){
		log.debug("livetest sendMessage1:");
        String msgStr = JSONObject.fromObject(msg).toString();
		log.debug("livetest sendMessage2:" + msgStr);
        TextMessage message = new TextMessage(msgStr);
		log.debug("livetest sendMessage:" + msgStr);
        try {
            this.session.sendMessage(message);
        } catch (IOException e) {
            log.error("send msg fail for:" + msgStr,e);
        }
    }
    /*
     * socket处理内部类
     */
    private class MyWebSocketHandler implements WebSocketHandler{
        @Override
        public void afterConnectionEstablished(WebSocketSession session) throws Exception {
            log.debug("afterConnectionEstablished in");
            String tenantCode = properties.getProperty("local.tenantCode");
            //发送socket登录请求
            LsMessage msg = new LsMessage();
            msg.setCode("01");
            msg.setTenantCode(tenantCode);
            String msgStr = JSONObject.fromObject(msg).toString();
            TextMessage message = new TextMessage(msgStr);
            //获取session作为全局变量
            SocketClient.this.session = session;
            session.sendMessage(message);  
        	conFlag="ready";
        }

        @Override
        public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
            log.debug("handleMessage in local");
            JSONObject jo = JSONObject.fromObject(message.getPayload());
            log.debug("------------handleMessage in local-------------- "+jo);
            //推流请求
            if(jo.getString("code").equals("2001")){
        		log.debug("livetest "+2001 );
        		pool.submit(() -> {
        			// 接收到推流成功后，进行云端拉流尝试，拉流成功后再通知云端业务服务器
                    try {
//						videoProcessor.startRemoteLive(session,jo.getString("businessId"),jo.getString("deviceCode"));
						videoProcessor.startRemoteLive(jo.getString("businessId"),jo.getString("deviceCode"));
					} catch (Exception e) {
			            log.debug(" handleMessage-------------- "+e);
					}
                    //return;
        		});

            }
            //停止推流
            if(jo.getString("code").equals("3001")){
        		log.debug("--------------------------------stopRemoteLive 0" +3001);
//                videoProcessor.stopRemoteLive(session,jo.getString("businessId"),jo.getString("deviceCode"));
                videoProcessor.stopRemoteLive(jo.getString("businessId"),jo.getString("deviceCode"));
            }
        }

        @Override
        public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
            // TODO Auto-generated method stub
            
        }

        @Override
        public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        	conFlag="none";
			while ("none".equals(conFlag)) {
        		log.debug("------afterConnectionClosed  reconnect " +conFlag);
				connect2();
				Thread.sleep(5000);
			}   
        }

        @Override
        public boolean supportsPartialMessages() {
            // TODO Auto-generated method stub
            return false;
        }
        
    }
    /*
     * socket处理内部类
     */
    private class MyStompSessionHandler extends StompSessionHandlerAdapter{
        private String tenantCode;//对应租户编码
    
        public MyStompSessionHandler(String tenantCode){
            this.tenantCode = tenantCode;
        }
    
        private void showHeaders(StompHeaders headers) {
            for (Map.Entry<String, List<String>> e : headers.entrySet()) {
                log.error("  " + e.getKey() + ": ");
                boolean first = true;
                for (String v : e.getValue()) {
                    if (!first)
                        System.err.print(", ");
                    log.error("fail v:" + v);
                    first = false;
                }
                log.error("show header fail");
            }
        }
    
        private void sendJsonMessage(StompSession session){
            LsMessage msg = new LsMessage();
            session.send("localServer", msg);
        }
    
        private void subscribeTopic(String topic,StompSession session){
            session.subscribe(topic, new StompFrameHandler() {
    
                @Override
                public Type getPayloadType(StompHeaders headers) {
                    return LsMessage.class;
                }
    
                @Override
                public void handleFrame(StompHeaders headers,Object payload){
                    log.debug("msg payload in:" + payload.toString());
                }
            });
        }
    }
}
